package com.zhanghe.study.hbase;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * 测试hbaseApi
 *
 * @author zh
 * @date 2021/4/24 15:00
 */
public class TestApi {

    // 使用Admin对象来操作DDL
    private static Admin admin;
    // 使用Table对象来操作DML
    private static Table table;
    private static Table tableIncr;


    static {
        try {
            // 获取连接
            Connection conn = ConnectionFactory.createConnection();
            admin = conn.getAdmin();
            table = conn.getTable(TableName.valueOf("test"));
            tableIncr = conn.getTable(TableName.valueOf("counters"));
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断表是否存在
     *
     * @param tableName
     * @return
     * @throws IOException
     */
    public static boolean isExistsTable(String tableName) throws IOException {
        return admin.tableExists(TableName.valueOf(tableName));

    }

    /**
     * 创建表
     */
    public static void createTable(String tableName, String... cfs) throws IOException {
        if (StringUtils.isBlank(tableName)) {
            throw new RuntimeException("表名不能为空");
        }
        if (cfs == null || cfs.length < 1) {
            throw new RuntimeException("列族不能为空");
        }
        List<ColumnFamilyDescriptor> cfList = new ArrayList<>();
        for (String cfName : cfs) {
            ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.of(cfName);
            cfList.add(columnFamilyDescriptor);
        }

        TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(TableName.valueOf(tableName))
                .setColumnFamilies(cfList)
                .build();


        admin.createTable(tableDescriptor);
    }


    public static void main(String[] args) throws IOException {
        // 测试isExistsTable 方法
//        System.out.println(isExistsTable("aa"));

        testPut("1234","info","name","zs");
//        testDelete("123456","info","name");
//        testGet("123456","info","name");
//        testScanWithFilter("123456","info","name");

//        testIncr("did222", "camp", "501");

    }

    /**
     * 插入数据
     *
     * @param rowKey
     * @param family
     * @param q
     * @param value
     * @throws IOException
     */
    public static void testPut(String rowKey, String family, String q, String value) throws IOException {
        Put put = new Put(Bytes.toBytes(rowKey));

        put.addColumn(Bytes.toBytes(family), Bytes.toBytes(q), Bytes.toBytes(value));
        table.put(put);
    }

    /**
     * 获取数据
     *
     * @param rowKey
     * @param family
     * @param q
     * @throws IOException
     */
    public static void testGet(String rowKey, String family, String q) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));

        get.addColumn(Bytes.toBytes(family), Bytes.toBytes(q));

        Result result = table.get(get);
        byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(q));
        System.out.println(Bytes.toString(value));
    }

    /**
     * 删除
     *
     * @param rowKey
     * @param family
     * @param q
     * @throws IOException
     */
    public static void testDelete(String rowKey, String family, String q) throws IOException {
        Delete delete = new Delete(Bytes.toBytes(rowKey));

        delete.addColumn(Bytes.toBytes(family), Bytes.toBytes(q));
        table.delete(delete);
    }

    /**
     * scan
     *
     * @param rowKey
     * @param family
     * @param q
     * @throws IOException
     */
    public static void testScanWithFilter(String rowKey, String family, String q) throws IOException {
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes(family), Bytes.toBytes(q));
        // 分页过滤器，只取一行数据
        Filter filter = new PageFilter(1);
        scan.setFilter(filter);

        ResultScanner scanner = table.getScanner(scan);
        for (Result result : scanner) {
            byte[] value = result.getValue(Bytes.toBytes(family), Bytes.toBytes(q));
            System.out.println(Bytes.toString(value));
        }

    }

    /**
     * 计数器
     *
     * @param rowKey
     * @param family
     * @param q
     * @throws IOException
     */
    public static void testIncr(String rowKey, String family, String q) throws IOException {
        // 计数器，自增1
        tableIncr.incrementColumnValue(Bytes.toBytes(rowKey),
                Bytes.toBytes(family), Bytes.toBytes(q), 1);
        // 获取计数器当前值，不做自增
        long value = tableIncr.incrementColumnValue(Bytes.toBytes(rowKey),
                Bytes.toBytes(family), Bytes.toBytes(q), 0);
        System.out.println(value);
    }

}
